/*


Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package controllers

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"iop.inspur.com/util"
	"math/rand"
	"regexp"
	"strconv"
	"strings"
	"time"

	"github.com/sirupsen/logrus"
	"iop.inspur.com/common/version"
	"iop.inspur.com/constants"
	"k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/runtime/serializer"

	//	"fmt"
	"github.com/go-logr/logr"
	ckev1alpha1 "iop.inspur.com/api/v1alpha1"
	batchv1 "k8s.io/api/batch/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	ClusterFinalizerName = "finalizer.stable.cluster.cke.io"

	ScaleJobAppLabel        = "cke-scale-nodes-job"
	DownJobAppLabel         = "cke-down-nodes-job"
	PatchJobAppLabel        = "cke-patchs-job"
	addAddonsJobAppLabel    = "cke-addaddons-job"
	deleteAddonsJobAppLabel = "cke-deleteaddons-job"
	CkePatchCpJobAppLabel   = "cke-patchs-create-job"

	CompletedStatus = "Completed"
	RunningStatus   = "Running"
	FailedStatus    = "Failed"
	InstallStatus   = "Installing"
	UpgradeStatus   = "Upgrading"
	TerminateStatus = "Terminating"

	// cke patch status
	PendingStatus = "Pending"
	SkippedStatus = "Skipped"
)

var source *rand.Rand

const RequeueAfter = 5

func init() {
	source = rand.New(rand.NewSource(time.Now().Unix()))
}

// CkeClusterReconciler reconciles a CkeCluster object
type CkeClusterReconciler struct {
	client.Client
	Log    logr.Logger
	Scheme *runtime.Scheme
}

// TODO: 失败了如何重新开始
//  ||
//  V
// DONE ? 失败了会将版本号改回去，需要重新改回来进行升级

// +kubebuilder:rbac:groups=cke.inspur.com,resources=ckeclusters,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=cke.inspur.com,resources=ckeclusters/status,verbs=get;update;patch
func (r *CkeClusterReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
	logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, **************", req.NamespacedName)
	ctx := context.Background()
	var addonsChanged bool
	var requeue = false
	var err error

	logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, get CkeCluster.", req.NamespacedName)
	//get ckecluster
	ckecluster := &ckev1alpha1.CkeCluster{}
	if err := r.Get(ctx, req.NamespacedName, ckecluster); err != nil {
		if errors.IsNotFound(err) {
			logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, CkeCluster not found.", req.NamespacedName)
			return ctrl.Result{}, nil
		}
		logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, failed to get CkeCluster. error=%s", req.NamespacedName, err.Error())
		return ctrl.Result{}, err
	}
	logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, success to get CkeCluster. version=%s", req.NamespacedName, ckecluster.Spec.Version)

	logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, get CkeNodeList.", req.NamespacedName)

	//get ckenodes
	ckenodes := &ckev1alpha1.CkeNodeList{}
	if err = r.List(ctx, ckenodes); err != nil {
		if errors.IsNotFound(err) {
			requeue = true
			logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, CkeNodeList not found.", req.NamespacedName)
			return ResultIfRequeue(requeue), nil
		}
		logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, failed to get CkeNodeList. error=%s", req.NamespacedName, err.Error())
		return ctrl.Result{}, err
	}
	nodes := make([]string, 0)
	for _, ckenode := range ckenodes.Items {
		nodes = append(nodes, ckenode.Name)
	}
	logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, success to get CkeNodeList. CkeNodeList=%s", req.NamespacedName, strings.Join(nodes, ", "))

	//add finalizers to ckecluster
	if ckecluster.DeletionTimestamp.IsZero() {
		if !ContainsStrings(ckecluster.Finalizers, ClusterFinalizerName) {
			logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, update CkeCluster with Finalizers.", req.NamespacedName)
			ckecluster.Finalizers = append(ckecluster.Finalizers, ClusterFinalizerName)
			if err := r.Update(ctx, ckecluster); err != nil {
				logrus.Errorf("CkeCluster Reconcile, CkeCluster=%s, failed to update CkeCluster with Finalizers. error=%s", req.NamespacedName, err.Error())
				return ctrl.Result{}, err
			}
		}
	}

	//delete ckecluster
	if !ckecluster.DeletionTimestamp.IsZero() && ContainsStrings(ckecluster.Finalizers, ClusterFinalizerName) {
		logrus.Errorf("CkeCluster Reconcile, CkeCluster=%s, Not allowed to delete CkeCluster.", req.NamespacedName)
		ckecluster.DeletionTimestamp = nil
		ckecluster.Status.Reason = "Not allowed"
		ckecluster.Status.Message = "Use \"ckeadm delete\" to delete kubernetes cluster"
		if err = r.Update(ctx, ckecluster); err != nil {
			logrus.Errorf("CkeCluster Reconcile, CkeCluster=%s, failed to update CkeCluster with 'Not allowed' message. error=%s", req.NamespacedName, err.Error())
			return ctrl.Result{}, err
		}
	}

	//init ckecluster
	if ckecluster.Status.State == constants.EmptyString {
		logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, update CkeCluster with 'Running' state.", req.NamespacedName)
		ckecluster.Status.State = RunningStatus
		err := r.Update(ctx, ckecluster)
		if err != nil {
			logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, failed to update CkeCluster with 'Running' state. error=%s", req.NamespacedName, err.Error())
			return ctrl.Result{}, err
		}
	}
	//检查集群status中是否有版本，若没有版本，把版本置为spec中的版本
	if ckecluster.Status.Version == "" {
		// 如果 version 为空，删除 annotation 中的 kubectl 标记，防止下次 apply status 被覆盖
		delete(ckecluster.Annotations, corev1.LastAppliedConfigAnnotation)

		logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, update CkeCluster with 'SpecVersion' status.", req.NamespacedName)

		ckecluster.Status.Version = ckecluster.Spec.Version
		err := r.Update(ctx, ckecluster)
		if err != nil {
			logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, failed to update CkeCluster with 'SpecVersion' status. error=%s", req.NamespacedName, err.Error())
			return ctrl.Result{}, err
		}
	}

	//install addons when not use component
	if len(ckecluster.Spec.ClusterVips) != 0 {
		// 若集群中 addons 未初始化，则初始化 addons
		if !ckecluster.Status.Initialization {
			logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, update CkeCluster with RunningAddons.", req.NamespacedName)
			ckecluster.Status.RunningAddons = []string{}
			for _, addon := range ckecluster.Spec.Addons {
				ckecluster.Status.RunningAddons = append(ckecluster.Status.RunningAddons, addon.Name)
			}
			ckecluster.Status.Initialization = true
			err := r.Update(ctx, ckecluster)
			if err != nil {
				logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, failed to update addons status. error=%s", req.NamespacedName, err.Error())
				return ctrl.Result{}, err
			}
		}

		//检查集群status中的addons和spec中的addons是否一致，若不一致，则置变量addonsChanged为true
		var statusAddons, specAddons []string
		statusAddons = ckecluster.Status.RunningAddons
		for _, specAddon := range ckecluster.Spec.Addons {
			specAddons = append(specAddons, specAddon.Name)
		}
		toAddAddons := Difference(specAddons, statusAddons)
		toDeleteAddons := []string{} // Difference(statusAddons, specAddons)
		if len(toAddAddons) != 0 || len(toDeleteAddons) != 0 {
			addonsChanged = true
		}

		//addons有变化
		if addonsChanged {
			logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, addons Changed.", req.NamespacedName)

			requeue = true
			//检查集群status中的状态是否为Running，若不为Running状态，则返回
			if ckecluster.Status.State != RunningStatus {
				return ctrl.Result{Requeue: true, RequeueAfter: RequeueAfter * 2 * time.Second}, nil
			}

			logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, get JobList of 'cke-add-addons-job'.", req.NamespacedName)

			//查找集群中目前的存在的addons job
			var CkeJobs = &batchv1.JobList{}
			var JobSelector = map[string]string{
				"app":       addAddonsJobAppLabel,
				"clusterID": ckecluster.Name,
			}

			if err := r.List(ctx, CkeJobs, client.MatchingLabels(JobSelector)); err != nil {
				logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, failed to get JobList of 'cke-add-addons-job'. error=%s", req.NamespacedName, err.Error())
				return ctrl.Result{}, err
			}
			if len(CkeJobs.Items) > 0 {
				for item, job := range CkeJobs.Items {
					if JobCompleted(&job) {
						logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, cke-add-addons-job completed; update CkeCluster with RunningAddons.", req.NamespacedName)
						ckecluster.Status.RunningAddons = append(ckecluster.Status.RunningAddons, job.ObjectMeta.Annotations[constants.AddonsAnnotationsName])
						if err := r.Update(ctx, ckecluster); err != nil {
							logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, cke-add-addons-job completed; failed to update CkeCluster with RunningAddons. error=%s", req.NamespacedName, err.Error())
							return ctrl.Result{}, err
						}
						logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, cke-add-addons-job completed; delete cke-add-addons-job.", req.NamespacedName)
						if err := r.Delete(ctx, &CkeJobs.Items[item], &constants.BackGroundDeleted); err != nil {
							logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, cke-add-addons-job completed; delete cke-add-addons-job. error=%s", req.NamespacedName, err.Error())
							return ctrl.Result{}, err
						}
					} else if JobFailed(&job) {
						ckecluster.Status.Reason = fmt.Sprintf("addons(%s) failed", job.Name)
						ckecluster.Status.Message = fmt.Sprintf("can't install addons(%s)", job.Name)
						for k, addon := range ckecluster.Spec.Addons {
							if addon.Name == job.ObjectMeta.Annotations[constants.AddonsAnnotationsName] {
								ckecluster.Spec.Addons = append(ckecluster.Spec.Addons[:k], ckecluster.Spec.Addons[k+1:]...)
								break
							}
						}
						logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, cke-add-addons-job failed; update CkeCluster with 'failed to install addons' message.", req.NamespacedName)
						ckecluster.Status.Message = fmt.Sprintf("fail to install addons(%s)", job.ObjectMeta.Annotations[constants.AddonsAnnotationsName])
						if err := r.Update(ctx, ckecluster); err != nil {
							logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, cke-add-addons-job failed; failed to update CkeCluster with 'failed to install addons' message. error=%s", req.NamespacedName, req.NamespacedName)
							return ctrl.Result{}, err
						}
						logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, cke-add-addons-job failed; delete cke-add-addons-job.", req.NamespacedName)
						if err := r.Delete(ctx, &CkeJobs.Items[item], &constants.BackGroundDeleted); err != nil {
							logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, cke-add-addons-job failed; delete cke-add-addons-job. error=%s", req.NamespacedName, err.Error())
							return ctrl.Result{}, err
						}
					}
				}
				return ctrl.Result{Requeue: true, RequeueAfter: RequeueAfter * time.Second}, nil
			}

			for _, toAddAddon := range toAddAddons {
				logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, create cke-add-addons-job. addons=%s", req.NamespacedName, toAddAddon)
				if err := r.Create(ctx, newaddAddonsJob(ckecluster, ckenodes, toAddAddon)); err != nil {
					logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, failed to create cke-del-addons-job. addons=%s error=%s", req.NamespacedName, toAddAddon, err.Error())
					return ctrl.Result{}, err
				}
			}
			for _, todeleteAddon := range toDeleteAddons {
				logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, create cke-add-addons-job. addons=%s", req.NamespacedName, todeleteAddon)
				if err := r.Create(ctx, newdeleteAddonsJob(ckecluster, ckenodes, todeleteAddon)); err != nil {
					logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, failed to create cke-del-addons-job. addons=%s error=%s", req.NamespacedName, todeleteAddon, err.Error())
					return ctrl.Result{}, err
				}
			}
			return ResultIfRequeue(requeue), nil
		}
	}

	//upgrade ckecluster
	if ckecluster.Status.Version != ckecluster.Spec.Version {
		//检查集群status中的状态是否为Running，若不为Running或者Upgrading状态，则返回
		if ckecluster.Status.State != RunningStatus && ckecluster.Status.State != UpgradeStatus {
			logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, Status.State is not running or upgrading. requeue. state=%s", req.NamespacedName, ckecluster.Status.State)
			return ctrl.Result{Requeue: true, RequeueAfter: RequeueAfter * 2 * time.Second}, nil
		}
		//检查集群status中的version和spec中的version是否一致，若一致，则返回
		if ckecluster.Status.Version == ckecluster.Spec.Version {
			logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, update CkeCluster with 'Running' status.", req.NamespacedName)
			ckecluster.Status.State = RunningStatus
			if err := r.Update(ctx, ckecluster); err != nil {
				logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, failed to update CkeCluster with 'Running' status. error=%s", req.NamespacedName, err.Error())
				return ctrl.Result{}, err
			}
			return ctrl.Result{}, nil
		}

		logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, update CkeCluster with 'Upgrading' status.", req.NamespacedName)

		// 版本不一致设置为升级状态
		ckecluster.Status.State = UpgradeStatus
		if err := r.Update(ctx, ckecluster); err != nil {
			logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, failed to update CkeCluster with 'Upgrading' status. error=%s", req.NamespacedName, err.Error())
			return ctrl.Result{}, err
		}

		//集群为运行中状态，且status中的version和spec中的version不一致，需要升级
		//获取现有patch cr
		logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, get CkePatchList.", req.NamespacedName)
		ckePatchList := &ckev1alpha1.CkePatchList{}
		err := r.List(ctx, ckePatchList)
		if err != nil {
			logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, failed to get patchList. error=%s", req.NamespacedName, err.Error())
			return ctrl.Result{}, err
		}

		isMatchFromVersionCkepatchExist := false
		isMatchToVersionCkepatchExist := false
		for _, ckePatch := range ckePatchList.Items {
			if regexp.MustCompile(fmt.Sprintf("^%s$", ckePatch.Spec.FromVersion)).MatchString(ckecluster.Status.Version) {
				isMatchFromVersionCkepatchExist = true
			}
			if ckePatch.Spec.ToVersion == ckecluster.Spec.Version {
				isMatchToVersionCkepatchExist = true
			}
		}

		//需要的ckepatch不存在时自动创建
		if !(isMatchFromVersionCkepatchExist && isMatchToVersionCkepatchExist) {
			logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, get CkePatch.", req.NamespacedName)
			requeue, err = r.GetCkePatch(ctx, ckecluster, ckePatchList)
			if err != nil {
				logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, failed to get CkePatch. error=%s", req.NamespacedName, err.Error())
				return ctrl.Result{}, err
			}
			return ResultIfRequeue(requeue), nil
		}

		//获取升级patch列表（返回patches中的名字，按顺序写到一个数组中）
		logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, get toExecutePatchList and toExecutePatchCr.", req.NamespacedName)
		var toExecutePatchList []string
		var toExecutePatchCr *ckev1alpha1.CkePatch
		toExecutePatchList, toExecutePatchCr, err = GetPatchList(ckecluster, ckePatchList)
		if err != nil {
			logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, failed to get toExecutePatchList and toExecutePatchCr. error=%s", req.NamespacedName, err.Error())
			logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, update CkeCluster with 'failed to get toExecutePatchList and toExecutePatchCr' message.", req.NamespacedName)
			ckecluster.Status.Reason = fmt.Sprintf("upgrade to version(%s) failed", ckecluster.Spec.Version)
			ckecluster.Status.Message = fmt.Sprintf("error while get upgrade patches, get error(%s)", err.Error())
			ckecluster.Spec.Version = ckecluster.Status.Version
			ckecluster.Status.ActivePatchJob = ""
			ckecluster.Status.State = RunningStatus
			if err := r.Update(ctx, ckecluster); err != nil {
				logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, failed to update CkeCluster with 'failed to get toExecutePatchList and toExecutePatchCr' message. error=%s", req.NamespacedName, err.Error())
				return ctrl.Result{}, err
			}
			return ctrl.Result{}, nil
		}

		// patch 包为空时表示执行完成
		if ckecluster.Status.ActivePatchJob == "" && len(toExecutePatchList) == 0 && len(toExecutePatchCr.Spec.ComponentPatches) == len(toExecutePatchCr.Status.CompletedComponentPatches) {
			ckecluster.Status.ActivePatchJob = ""
			ckecluster.Status.ActiveCKEPatch = ""
			ckecluster.Status.Message = ""
			ckecluster.Status.Reason = ""
			ckecluster.Status.Version = toExecutePatchCr.Spec.ToVersion
			ckecluster.Status.State = RunningStatus
			ckecluster.Status.CompletedCKEPatchs = AppendIfNotExist(ckecluster.Status.CompletedCKEPatchs, toExecutePatchCr.Name)

			logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, delete all of 'cke-patch-job'", req.NamespacedName)
			r.DeleteAllOf(ctx, &batchv1.Job{}, client.InNamespace(req.Namespace), &client.DeleteAllOfOptions{DeleteOptions: constants.BackGroundDeleted}, client.MatchingLabels(map[string]string{
				"app": PatchJobAppLabel,
			})) // deleted cke patch job after completed

			logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, update CkeCluster with 'Running' status.", req.NamespacedName)
			if err := r.Update(ctx, ckecluster); err != nil {
				logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, failed to update CkeCluster with 'Running' status. error=%s", req.NamespacedName, err.Error())
				return ctrl.Result{}, err
			}
			logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, requeue. *****", req.NamespacedName)
			return ctrl.Result{Requeue: true, RequeueAfter: RequeueAfter * time.Second}, nil
		}

		// 执行组件更新
		if len(toExecutePatchCr.Spec.ComponentPatches) > len(toExecutePatchCr.Status.FailedComponentPatches)+len(toExecutePatchCr.Status.CompletedComponentPatches) {
			logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, UpdateAndCheckCkeComponents.", req.NamespacedName)
			err = r.UpdateAndCheckCkeComponents(ctx, toExecutePatchCr, ckecluster)
			if err != nil {
				if strings.Contains(err.Error(), "just return") {
					return ctrl.Result{}, nil
				}
				logrus.Errorf("CkeCluster Reconcile, CkeCluster=%s, failed to UpdateAndCheckCkeComponents. error=%s", req.NamespacedName, err.Error())
				return ctrl.Result{}, err
			}
			requeue = true
		}

		//启动升级job
		//查看是否有activejob,若没有则按顺序启动job，若有，获取该job，检查是否完成，没完成则返回并重新放入队列，
		//若已经完成,，则取列表中下一个patch启动一个job并更新activeJob
		//若已完成的是最后一个patch将cluster version更新，将activejob去除，返回并重新放入队列
		if ckecluster.Status.ActivePatchJob != "" {
			logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, get JobList of 'cke-patch-job'.", req.NamespacedName)

			var CkeJobs = &batchv1.JobList{}
			var JobSelector = map[string]string{
				"app":         PatchJobAppLabel,
				"version":     toExecutePatchCr.Spec.ToVersion,
				"patchSerial": fmt.Sprintf("b%se", StrNoLongThan(base64.RawURLEncoding.EncodeToString([]byte(ckecluster.Status.ActivePatchJob)), constants.AnnotationLengthLimit-2)),
			}

			if err := r.List(ctx, CkeJobs, client.MatchingLabels(JobSelector)); err != nil {
				logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, failed to get JobList of 'cke-patch-job'. error=%s", req.NamespacedName, err.Error())
				return ctrl.Result{}, err
			}
			for item, job := range CkeJobs.Items {
				if !JobCompleted(&job) && !JobFailed(&job) {
					logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, cke-patch-job is running. requeue. ********  job=%s", req.NamespacedName, job.Name)
					return ctrl.Result{Requeue: true, RequeueAfter: RequeueAfter * time.Second}, nil
				}
				logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, cke-patch-job is completed or failed. delete cke-patch-job. job=%s", req.NamespacedName, job.Name)
				if err := r.Delete(ctx, &CkeJobs.Items[item], &constants.BackGroundDeleted); err != nil {
					logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, cke-patch-job is completed or failed. failed to delete cke-patch-job. job=%s error=%s", req.NamespacedName, job.Name, err.Error())
					return ctrl.Result{}, err
				}
			}
			for _, job := range CkeJobs.Items {
				if JobCompleted(&job) { // 成功时标记已经完成 job
					toExecutePatchCr.Status.CompletedPatchs = AppendIfNotExist(toExecutePatchCr.Status.CompletedPatchs, ckecluster.Status.ActivePatchJob)
					if err := r.Update(ctx, toExecutePatchCr); err != nil {
						logrus.Infof("fail to update ckepatch (%s) status information", toExecutePatchCr.Name)
						return ctrl.Result{}, err
					}
				}
				if JobFailed(&job) { // failed 重新设置版本 退出
					if len(toExecutePatchCr.Spec.ComponentPatches) > len(toExecutePatchCr.Status.FailedComponentPatches)+len(toExecutePatchCr.Status.CompletedComponentPatches) {
						return ctrl.Result{Requeue: true, RequeueAfter: RequeueAfter * time.Second}, nil
					}
					ckecluster.Spec.Version = ckecluster.Status.Version
					ckecluster.Status.Reason = fmt.Sprintf("version(%s) patch(%s) failed", toExecutePatchCr.Spec.ToVersion, ckecluster.Status.ActivePatchJob)
					ckecluster.Status.Message = fmt.Sprintf("can't upgrade to version(%s), change to current version(%s)", ckecluster.Spec.Version, ckecluster.Status.Version)
					ckecluster.Status.ActivePatchJob = ""
					ckecluster.Status.State = RunningStatus
					if err := r.Update(ctx, ckecluster); err != nil {
						return ctrl.Result{}, err
					}
					return ctrl.Result{}, nil
				}
			}

			if !strings.Contains(ckecluster.Status.Message, "some patches are still running") {
				ckecluster.Status.Reason = ""
				ckecluster.Status.Message = ""
				if err := r.Update(ctx, ckecluster); err != nil {
					return ctrl.Result{}, err
				}
			}
			// 没有失败且在运行的 job 时，设置当前 activePatch 为空，等待下个 patch 包执行
			ckecluster.Status.ActivePatchJob = ""
			if err := r.Update(ctx, ckecluster); err != nil {
				return ctrl.Result{}, err
			}
			return ctrl.Result{Requeue: true}, nil
		} else {
			if len(toExecutePatchList) > 0 {
				//还没开始执行或者已经升到了中间版本，按顺序执行 toExecutePtchList
				var execPatch = &PatchInformation{
					Name:        toExecutePatchList[0],
					ImageName:   toExecutePatchCr.Spec.Image,
					CurrVersion: toExecutePatchCr.Spec.ToVersion,
					ToVersion:   ckecluster.Spec.Version,
				}
				r.DeleteAllOf(ctx, &batchv1.Job{}, client.InNamespace(req.Namespace), &client.DeleteAllOfOptions{DeleteOptions: constants.BackGroundDeleted}, client.MatchingLabels(map[string]string{
					"app": PatchJobAppLabel,
				}))
				if err := r.Create(ctx, newPatchJob(ckecluster, ckenodes, execPatch)); err != nil {
					logrus.Errorf("fail to create patch job for ckecluster(%s), patch(%s), get error: %s", ckecluster.Name, execPatch.Name, err.Error())
					return ctrl.Result{}, err
				}
				ckecluster.Status.ActiveCKEPatch = toExecutePatchCr.Name
				ckecluster.Status.ActivePatchJob = execPatch.Name
				err := r.Update(ctx, ckecluster)
				if err != nil {
					logrus.Infof("CkeCluster Reconcile, CkeCluster=%s, cluster update state error! err: %s", err.Error())
					return ctrl.Result{}, err
				}
				return ctrl.Result{Requeue: true, RequeueAfter: RequeueAfter * time.Second}, nil
			}
		}
	}

	return ResultIfRequeue(requeue), nil
}

func (r *CkeClusterReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&ckev1alpha1.CkeCluster{}).
		Complete(r)
}

func (r *CkeClusterReconciler) GetCkePatch(ctx context.Context, ckecluster *ckev1alpha1.CkeCluster, ckePatchList *ckev1alpha1.CkePatchList) (bool, error) {
	var err error
	var requeue = false
	var CkeJobs = &batchv1.JobList{}
	var JobSelector = map[string]string{
		"app":       CkePatchCpJobAppLabel,
		"clusterID": ckecluster.Name,
	}
	if err := r.List(ctx, CkeJobs, client.MatchingLabels(JobSelector)); err != nil {
		return requeue, err
	}
	if len(CkeJobs.Items) == 0 {
		if err = r.Create(ctx, newCpCkePatchJob(ckecluster)); err != nil {
			logrus.Errorf("CkeCluster Reconcile, CkeCluster=%s, newCpCkePatchJob error! err: %s", err.Error())
			return requeue, err
		}
		logrus.Infof("success create job for cp ckepatchs")
		requeue = true
		return requeue, nil
	} else {
		if JobCompleted(&CkeJobs.Items[0]) {
			if strings.Contains(ckecluster.Status.Message, "while cp ckepatchs") {
				ckecluster.Status.Reason = ""
				ckecluster.Status.Message = ""
				if err := r.Update(ctx, ckecluster); err != nil {
					return requeue, err
				}
			}
			err = r.CreateCkePatch(ctx, ckePatchList, ckecluster.Status.Version, ckecluster.Spec.Version)
			if err != nil {
				logrus.Errorf("CkeCluster Reconcile, CkeCluster=%s, CreateCkePatch error! err: %s", err.Error())
				ckecluster.Status.Reason = fmt.Sprintf("upgrade to version(%s) failed", ckecluster.Spec.Version)
				ckecluster.Status.Message = fmt.Sprintf("error while create ckepatches, get error(%s)", err.Error())
				ckecluster.Spec.Version = ckecluster.Status.Version
				ckecluster.Status.State = RunningStatus
				if err := r.Update(ctx, ckecluster); err != nil {
					return requeue, err
				}
				return requeue, err
			}
			if strings.Contains(ckecluster.Status.Message, "while create ckepatches") {
				ckecluster.Status.Reason = ""
				ckecluster.Status.Message = ""
				if err := r.Update(ctx, ckecluster); err != nil {
					return requeue, err
				}
			}
			if err = r.Delete(ctx, &CkeJobs.Items[0], &constants.BackGroundDeleted); err != nil {
				logrus.Errorf("CkeCluster Reconcile, CkeCluster=%s, delete job error! err: %s", err.Error())
				return requeue, err
			}
			requeue = true
			return requeue, nil
		}
		if JobFailed(&CkeJobs.Items[0]) {
			if err = r.Delete(ctx, &CkeJobs.Items[0], &constants.BackGroundDeleted); err != nil {
				logrus.Errorf("CkeCluster Reconcile, CkeCluster=%s, delete job error! err: %s", err.Error())
				return requeue, err
			}
			ckecluster.Status.Reason = fmt.Sprintf("upgrade to version(%s) failed", ckecluster.Spec.Version)
			ckecluster.Status.Message = fmt.Sprintf("error while cp ckepatchs")
			ckecluster.Spec.Version = ckecluster.Status.Version
			ckecluster.Status.State = RunningStatus
			if err := r.Update(ctx, ckecluster); err != nil {
				return requeue, err
			}
			return requeue, nil
		}
		requeue = true
		return requeue, nil
	}
}

func (r *CkeClusterReconciler) CreateCkePatch(ctx context.Context, ckepatchlist *ckev1alpha1.CkePatchList, fromVersion string, toVersion string) error {
	isMatchFromVersionCkepatchExist := false
	isMatchToVersionCkepatchExist := false

	ckepatchs, err := util.GetCkePatch(CkeControllerCkePatchPath)
	if err != nil {
		return err
	}
	for _, ckepatch := range *ckepatchs {
		if regexp.MustCompile(fmt.Sprintf("^%s$", ckepatch.Spec.FromVersion)).MatchString(fromVersion) {
			isMatchFromVersionCkepatchExist = true
		}
		if ckepatch.Spec.ToVersion == toVersion {
			isMatchToVersionCkepatchExist = true
		}
	}
	if isMatchFromVersionCkepatchExist == false {
		return fmt.Errorf("no match fromVersion ckepatch exist")
	}
	if isMatchToVersionCkepatchExist == false {
		return fmt.Errorf("no match toVersion ckepatch exist")
	}

	createdCkepatchs := make(map[string]bool, len(ckepatchlist.Items))
	for _, createdCkepatch := range ckepatchlist.Items {
		createdCkepatchs[createdCkepatch.Name] = true
	}
	for _, ckepatch := range *ckepatchs {
		if &ckepatch != nil {
			if _, ok := createdCkepatchs[ckepatch.Name]; !ok {
				if err = r.Create(ctx, &ckepatch); err != nil {
					logrus.Errorf("CkeCluster Reconcile, CkeCluster=%s, create CkePatch error! err: %s", err.Error())
					return err
				}
			}
		}
	}

	return err
}

func (r *CkeClusterReconciler) UpdateAndCheckCkeComponents(ctx context.Context, toExecutePatchCr *ckev1alpha1.CkePatch, ckecluster *ckev1alpha1.CkeCluster) error {
	var err error

	ckecomponents := &ckev1alpha1.CkeComponentList{}
	if err = r.List(ctx, ckecomponents); err != nil {
		if errors.IsNotFound(err) {
			logrus.Infof("UpdateAndCheckCkeComponents, CkeComponentList not found!")
			return err
		}
		logrus.Errorf("UpdateAndCheckCkeComponents, get CkeComponentList err! err: %s", err.Error())
		return err
	}
	logrus.Infof("get information ckecomponentlist")
	ckecomponentsMap := make(map[string]*ckev1alpha1.CkeComponent)
	for i, _ := range ckecomponents.Items {
		ckecomponentsMap[ckecomponents.Items[i].Name] = &ckecomponents.Items[i]
	}

	if len(toExecutePatchCr.Status.ActiveComponentPatches) == 0 {
		notRunningComponents, err := r.AddActiveCkeComponents(ctx, toExecutePatchCr, ckecomponentsMap)
		if len(notRunningComponents) != 0 {
			ckecluster.Status.Reason = fmt.Sprintf("upgrade to version(%s) failed", ckecluster.Spec.Version)
			ckecluster.Status.Message = fmt.Sprintf("ckecomponents %s are not running", strings.Join(notRunningComponents, ","))
			ckecluster.Spec.Version = ckecluster.Status.Version
			ckecluster.Status.State = RunningStatus
			if err = r.Update(ctx, ckecluster); err != nil {
				return err
			}
			return fmt.Errorf("just return")
		}
		if strings.Contains(ckecluster.Status.Message, "are not running") {
			ckecluster.Status.Reason = ""
			ckecluster.Status.Message = ""
			if err := r.Update(ctx, ckecluster); err != nil {
				return err
			}
		}
		if err != nil {
			return err
		}
		return err
	}

	needDeletes := make([]int, 0)
	for i, ActiveComponentPatch := range toExecutePatchCr.Status.ActiveComponentPatches {
		isComponentPatchCompleted, isComponentPatchFailed := CheckCkeComponent(ActiveComponentPatch, ckecomponentsMap[ActiveComponentPatch.Name])
		if isComponentPatchFailed {
			toExecutePatchCr.Status.FailedComponentPatches = append(toExecutePatchCr.Status.FailedComponentPatches, ActiveComponentPatch)
			needDeletes = append(needDeletes, i)
		}
		if isComponentPatchCompleted {
			toExecutePatchCr.Status.CompletedComponentPatches = append(toExecutePatchCr.Status.CompletedComponentPatches, ActiveComponentPatch)
			needDeletes = append(needDeletes, i)
		}
	}
	if len(needDeletes) != 0 {
		for n, needDelete := range needDeletes {
			toExecutePatchCr.Status.ActiveComponentPatches = append(toExecutePatchCr.Status.ActiveComponentPatches[:needDelete-n], toExecutePatchCr.Status.ActiveComponentPatches[needDelete-n+1:]...)
		}
		if err = r.Update(ctx, toExecutePatchCr); err != nil {
			logrus.Errorf("UpdateAndCheckCkeComponents, %s/%s Update error! err: %s", toExecutePatchCr.Namespace, toExecutePatchCr.Name, err.Error())
			return err
		}
		if len(toExecutePatchCr.Status.FailedComponentPatches) > 0 {
			ckecluster.Status.Reason = fmt.Sprintf("upgrade to version(%s) failed, and some patches are still running", ckecluster.Spec.Version)
			ckecluster.Status.Message = fmt.Sprintf("See \"kubectl get ckepatch\" for details.")
			if err := r.Update(ctx, ckecluster); err != nil {
				return err
			}
		}
	}

	return err
}

func (r *CkeClusterReconciler) AddActiveCkeComponents(ctx context.Context, toExecutePatchCr *ckev1alpha1.CkePatch, ckecomponentsMap map[string]*ckev1alpha1.CkeComponent) ([]string, error) {
	var err error

	componentsPatches := make([]ckev1alpha1.CkeComponentPatch, len(toExecutePatchCr.Spec.ComponentPatches))
	copy(componentsPatches, toExecutePatchCr.Spec.ComponentPatches)

	notRunningComponents := make([]string, 0)
	notExistComponents := make([]int, 0)
	for i, ComponentPatch := range componentsPatches {
		if _, ok := ckecomponentsMap[ComponentPatch.Name]; !ok {
			notExistComponents = append(notExistComponents, i)
		} else if ckecomponentsMap[ComponentPatch.Name].Status.State != ComponentRunningStatus {
			notRunningComponents = append(notRunningComponents, ComponentPatch.Name)
		}
	}
	if len(notRunningComponents) != 0 {
		return notRunningComponents, err
	}
	if len(notExistComponents) != 0 {
		for _, i := range notExistComponents {
			toExecutePatchCr.Status.CompletedComponentPatches = append(toExecutePatchCr.Status.CompletedComponentPatches, componentsPatches[i])
		}
		if err = r.Update(ctx, toExecutePatchCr); err != nil {
			logrus.Errorf("AddActiveCkeComponents, %s/%s Update error! err: %s", toExecutePatchCr.Namespace, toExecutePatchCr.Name, err.Error())
			return notRunningComponents, err
		}
		for n, notExistComponent := range notExistComponents {
			componentsPatches = append(componentsPatches[:notExistComponent-n], componentsPatches[notExistComponent-n+1:]...)
		}
	}

	for _, ComponentPatch := range componentsPatches {
		if ComponentPatch.ToVersion != "" {
			ckecomponentsMap[ComponentPatch.Name].Spec.Version = ComponentPatch.ToVersion
		}
		if ComponentPatch.ToImageTag != "" {
			ckecomponentsMap[ComponentPatch.Name].Spec.ImageTag = ComponentPatch.ToImageTag
		}
		if len(ComponentPatch.Parameters) != 0 {
			if len(ckecomponentsMap[ComponentPatch.Name].Spec.Parameters) == 0 {
				ckecomponentsMap[ComponentPatch.Name].Spec.Parameters = make(map[string]string)
			}
			for k, v := range ComponentPatch.Parameters {
				ckecomponentsMap[ComponentPatch.Name].Spec.Parameters[k] = v
			}
		}
		if err = r.Update(ctx, ckecomponentsMap[ComponentPatch.Name]); err != nil {
			logrus.Errorf("AddActiveCkeComponents, %s/%s Update error! err: %s", ckecomponentsMap[ComponentPatch.Name].Namespace, ckecomponentsMap[ComponentPatch.Name].Name, err.Error())
			return notRunningComponents, err
		}
	}

	toExecutePatchCr.Status.ActiveComponentPatches = componentsPatches
	if err = r.Update(ctx, toExecutePatchCr); err != nil {
		logrus.Errorf("AddActiveCkeComponents, %s/%s Update error! err: %s", toExecutePatchCr.Namespace, toExecutePatchCr.Name, err.Error())
		return notRunningComponents, err
	}

	return notRunningComponents, err
}

// kubectl get job -n kube-system --selector=app=cke-scale-nodes-job,clusterID=cluster_for_app,nodeName=mgt04
// kubectl get job -n kube-system --selector=app=cke-down-nodes-job,clusterID=cluster_for_app,nodeName=mgt04

func newScaleUpNodeJob(ckecluster *ckev1alpha1.CkeCluster, nodes *ckev1alpha1.CkeNodeList) *batchv1.Job {
	var Para, Back, Com, Model = int32(1), int32(2), int32(1), int32(0600)
	var ImageName = ImageGetFromConfig(&ckecluster.Spec)
	var CkeConfigContent = Base64Context(ckecluster, &nodes.Items)
	var node = GetNodeByRole(&nodes.Items, constants.ScaleRole)
	var fork = strconv.Itoa(constants.CkeNodeForks)
	var execYAML = constants.ScaleYamlFile

	// 为 master 转 slave 节点
	if no := GetNodeByRole(&nodes.Items, constants.MasterToSlaveRole); no != nil && node.Name == no.Name {
		execYAML = constants.MasterToSlaveYamlFile
	}

	if ckecluster == nil || nodes == nil || node == nil {
		return nil
	}

	return &batchv1.Job{
		ObjectMeta: metav1.ObjectMeta{
			Name:      fmt.Sprintf("%s-%s", "scale-nodes-job", RandomUID()),
			Namespace: ckecluster.Namespace,
			OwnerReferences: []metav1.OwnerReference{
				*metav1.NewControllerRef(ckecluster, schema.GroupVersionKind{Group: ckev1alpha1.GroupVersion.Group, Version: ckev1alpha1.GroupVersion.Version, Kind: "CkeCluster"}),
			},
			Labels: map[string]string{
				"app":       ScaleJobAppLabel,
				"nodeName":  node.Name,
				"clusterID": ckecluster.Name,
			},
		},
		Spec: batchv1.JobSpec{
			Parallelism:  &Para,
			BackoffLimit: &Back,
			Completions:  &Com,
			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Name: "scale-nodes-pods",
				},
				Spec: corev1.PodSpec{
					HostNetwork: true,
					NodeSelector: map[string]string{
						"beta.kubernetes.io/os":          "linux",
						"node-role.kubernetes.io/master": "true",
					},
					Tolerations: []corev1.Toleration{
						{
							Key:      "CriticalAddonsOnly",
							Operator: "Exists",
						},
						{
							Operator: "Exists",
							Effect:   "NoExecute",
						},
						{
							Operator: "Exists",
							Effect:   "NoSchedule",
						},
					},
					Containers: []corev1.Container{{
						Name:            "cke-scale-nodes",
						Image:           ImageName,
						ImagePullPolicy: corev1.PullIfNotPresent,
						Command: []string{
							"/bin/playbook",
							"--execyml",
							execYAML,
							"--config",
							CkeConfigContent,
							"--fork",
							fork,
							"--logpath",
							constants.LogsInpodDir,
							"--logname",
							fmt.Sprintf("scale-%s", node.Name),
						},
						Env: []corev1.EnvVar{
							{
								Name:  "ANSIBLE_HOST_KEY_CHECKING",
								Value: "Flase",
							},
							{
								Name:  "ANSIBLE_GATHERING",
								Value: "explicit",
							},
						},
						VolumeMounts: []corev1.VolumeMount{
							{
								MountPath: "/root/.ssh/",
								Name:      "cke-certs",
							},
							{
								MountPath: constants.LogsInpodDir,
								Name:      "ckelogs",
							},
						},
					}},
					RestartPolicy: corev1.RestartPolicyOnFailure,
					Volumes: []corev1.Volume{
						{
							Name: "cke-certs",
							VolumeSource: corev1.VolumeSource{
								Secret: &corev1.SecretVolumeSource{
									DefaultMode: &Model,
									SecretName:  "cke-certs",
								},
							},
						},
						{
							Name: "ckelogs",
							VolumeSource: corev1.VolumeSource{
								HostPath: &corev1.HostPathVolumeSource{
									Path: constants.ScaleLogDir,
									Type: &constants.MountPathDirType,
								},
							},
						},
					},
				},
			},
		},
	}
}

func newScaleDownNodeJob(ckecluster *ckev1alpha1.CkeCluster, nodes *ckev1alpha1.CkeNodeList) *batchv1.Job {
	var TTL, Para, Back, Com, Model = int32(300), int32(1), int32(2), int32(1), int32(0600)
	var ImageName = ImageGetFromConfig(&ckecluster.Spec)
	var CkeConfigContent = Base64Context(ckecluster, &nodes.Items)
	var node = GetNodeByRole(&nodes.Items, "scale-down")
	var fork = strconv.Itoa(constants.CkeNodeForks)

	if ckecluster == nil || nodes == nil || node == nil {
		return nil
	}

	return &batchv1.Job{
		ObjectMeta: metav1.ObjectMeta{
			Name:      fmt.Sprintf("%s-%s", "down-nodes-job", RandomUID()),
			Namespace: ckecluster.Namespace,
			OwnerReferences: []metav1.OwnerReference{
				*metav1.NewControllerRef(ckecluster, schema.GroupVersionKind{Group: ckev1alpha1.GroupVersion.Group, Version: ckev1alpha1.GroupVersion.Version, Kind: "CkeCluster"}),
			},
			Labels: map[string]string{
				"app":       DownJobAppLabel,
				"nodeName":  node.Name,
				"clusterID": ckecluster.Name,
			},
		},
		Spec: batchv1.JobSpec{
			TTLSecondsAfterFinished: &TTL,
			Parallelism:             &Para,
			BackoffLimit:            &Back,
			Completions:             &Com,
			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Name: "down-nodes-pods",
				},
				Spec: corev1.PodSpec{
					HostNetwork: true,
					NodeSelector: map[string]string{
						"beta.kubernetes.io/os":          "linux",
						"node-role.kubernetes.io/master": "true",
					},
					Tolerations: []corev1.Toleration{
						{
							Key:      "CriticalAddonsOnly",
							Operator: "Exists",
						},
						{
							Operator: "Exists",
							Effect:   "NoExecute",
						},
						{
							Operator: "Exists",
							Effect:   "NoSchedule",
						},
					},
					Containers: []corev1.Container{{
						Name:            "cke-down-nodes",
						Image:           ImageName,
						ImagePullPolicy: corev1.PullIfNotPresent,
						Command: []string{
							"/bin/playbook",
							"--execyml",
							constants.DownYamlFile,
							"--config",
							CkeConfigContent,
							"--fork",
							fork,
							"--logpath",
							constants.LogsInpodDir,
							"--logname",
							fmt.Sprintf("down-%s", node.Name),
						},
						Env: []corev1.EnvVar{
							{
								Name:  "ANSIBLE_HOST_KEY_CHECKING",
								Value: "Flase",
							},
							{
								Name:  "ANSIBLE_GATHERING",
								Value: "explicit",
							},
						},
						VolumeMounts: []corev1.VolumeMount{
							{
								MountPath: "/root/.ssh/",
								Name:      "cke-certs",
							},
							{
								MountPath: constants.LogsInpodDir,
								Name:      "ckelogs",
							},
						},
					}},
					RestartPolicy: corev1.RestartPolicyOnFailure,
					Volumes: []corev1.Volume{
						{
							Name: "cke-certs",
							VolumeSource: corev1.VolumeSource{
								Secret: &corev1.SecretVolumeSource{
									DefaultMode: &Model,
									SecretName:  "cke-certs",
								},
							},
						},
						{
							Name: "ckelogs",
							VolumeSource: corev1.VolumeSource{
								HostPath: &corev1.HostPathVolumeSource{
									Path: constants.DownLogDir,
									Type: &constants.MountPathDirType,
								},
							},
						},
					},
				},
			},
		},
	}
}

func newaddAddonsJob(ckecluster *ckev1alpha1.CkeCluster, nodes *ckev1alpha1.CkeNodeList, addonName string) *batchv1.Job {
	var Para, Back, Com, Model = int32(1), int32(2), int32(1), int32(0600)
	var ImageName = ImageGetFromConfig(&ckecluster.Spec)
	var CkeConfigContent = Base64Context(ckecluster, &nodes.Items)
	var fork = strconv.Itoa(constants.ForkThreads)

	if ckecluster == nil || nodes == nil {
		return nil
	}

	return &batchv1.Job{
		ObjectMeta: metav1.ObjectMeta{
			Name:      fmt.Sprintf("%s-%s", "add-addons-job", RandomUID()),
			Namespace: ckecluster.Namespace,
			OwnerReferences: []metav1.OwnerReference{
				*metav1.NewControllerRef(ckecluster, schema.GroupVersionKind{Group: ckev1alpha1.GroupVersion.Group, Version: ckev1alpha1.GroupVersion.Version, Kind: "CkeCluster"}),
			},
			Annotations: map[string]string{
				constants.AddonsAnnotationsName: addonName,
			},
			Labels: map[string]string{
				"app":       addAddonsJobAppLabel,
				"clusterID": ckecluster.Name,
			},
		},
		Spec: batchv1.JobSpec{
			Parallelism:  &Para,
			BackoffLimit: &Back,
			Completions:  &Com,
			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Name: addAddonsJobAppLabel,
				},
				Spec: corev1.PodSpec{
					HostNetwork: true,
					NodeSelector: map[string]string{
						"beta.kubernetes.io/os":          "linux",
						"node-role.kubernetes.io/master": "true",
					},
					Tolerations: []corev1.Toleration{
						{
							Key:      "CriticalAddonsOnly",
							Operator: "Exists",
						},
						{
							Operator: "Exists",
							Effect:   "NoExecute",
						},
						{
							Operator: "Exists",
							Effect:   "NoSchedule",
						},
					},
					Containers: []corev1.Container{{
						Name:            addAddonsJobAppLabel,
						Image:           ImageName,
						ImagePullPolicy: corev1.PullIfNotPresent,
						Command: []string{
							"/bin/playbook",
							"--execyml",
							fmt.Sprintf("addons/%s.yml", addonName),
							"--config",
							CkeConfigContent,
							"--fork",
							fork,
							"--logpath",
							constants.LogsInpodDir,
							"--logname",
							fmt.Sprintf("addons-add-%s", addonName),
						},
						Env: []corev1.EnvVar{
							{
								Name:  "ANSIBLE_HOST_KEY_CHECKING",
								Value: "Flase",
							},
							{
								Name:  "ANSIBLE_GATHERING",
								Value: "explicit",
							},
						},
						VolumeMounts: []corev1.VolumeMount{
							{
								MountPath: "/root/.ssh/",
								Name:      "cke-certs",
							},
							{
								MountPath: constants.LogsInpodDir,
								Name:      "ckelogs",
							},
						},
					}},
					RestartPolicy: corev1.RestartPolicyOnFailure,
					Volumes: []corev1.Volume{
						{
							Name: "cke-certs",
							VolumeSource: corev1.VolumeSource{
								Secret: &corev1.SecretVolumeSource{
									DefaultMode: &Model,
									SecretName:  "cke-certs",
								},
							},
						},
						{
							Name: "ckelogs",
							VolumeSource: corev1.VolumeSource{
								HostPath: &corev1.HostPathVolumeSource{
									Path: constants.AddonsLogDir,
									Type: &constants.MountPathDirType,
								},
							},
						},
					},
				},
			},
		},
	}
}

func newdeleteAddonsJob(ckecluster *ckev1alpha1.CkeCluster, nodes *ckev1alpha1.CkeNodeList, addonName string) *batchv1.Job {
	var Para, Back, Com, Model = int32(1), int32(2), int32(1), int32(0600)
	var ImageName = ImageGetFromConfig(&ckecluster.Spec)
	var CkeConfigContent = Base64Context(ckecluster, &nodes.Items)
	var fork = strconv.Itoa(constants.ForkThreads)

	if ckecluster == nil || nodes == nil {
		return nil
	}

	return &batchv1.Job{
		ObjectMeta: metav1.ObjectMeta{
			Name:      fmt.Sprintf("%s-%s", "delete-addons-job", RandomUID()),
			Namespace: ckecluster.Namespace,
			OwnerReferences: []metav1.OwnerReference{
				*metav1.NewControllerRef(ckecluster, schema.GroupVersionKind{Group: ckev1alpha1.GroupVersion.Group, Version: ckev1alpha1.GroupVersion.Version, Kind: "CkeCluster"}),
			},
			Annotations: map[string]string{
				constants.AddonsAnnotationsName: addonName,
			},
			Labels: map[string]string{
				"app":       deleteAddonsJobAppLabel,
				"clusterID": ckecluster.Name,
			},
		},
		Spec: batchv1.JobSpec{
			Parallelism:  &Para,
			BackoffLimit: &Back,
			Completions:  &Com,
			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Name: deleteAddonsJobAppLabel,
				},
				Spec: corev1.PodSpec{
					HostNetwork: true,
					NodeSelector: map[string]string{
						"beta.kubernetes.io/os":          "linux",
						"node-role.kubernetes.io/master": "true",
					},
					Tolerations: []corev1.Toleration{
						{
							Key:      "CriticalAddonsOnly",
							Operator: "Exists",
						},
						{
							Operator: "Exists",
							Effect:   "NoExecute",
						},
						{
							Operator: "Exists",
							Effect:   "NoSchedule",
						},
					},
					Containers: []corev1.Container{{
						Name:            deleteAddonsJobAppLabel,
						Image:           ImageName,
						ImagePullPolicy: corev1.PullIfNotPresent,
						Command: []string{
							"/bin/playbook",
							"--execyml",
							fmt.Sprintf("addons/%s-delete.yml", addonName),
							"--config",
							CkeConfigContent,
							"--fork",
							fork,
							"--logpath",
							constants.LogsInpodDir,
							"--logname",
							fmt.Sprintf("addons-del-%s", addonName),
						},
						Env: []corev1.EnvVar{
							{
								Name:  "ANSIBLE_HOST_KEY_CHECKING",
								Value: "Flase",
							},
							{
								Name:  "ANSIBLE_GATHERING",
								Value: "explicit",
							},
						},
						VolumeMounts: []corev1.VolumeMount{
							{
								MountPath: "/root/.ssh/",
								Name:      "cke-certs",
							},
							{
								MountPath: constants.LogsInpodDir,
								Name:      "ckelogs",
							},
						},
					}},
					RestartPolicy: corev1.RestartPolicyOnFailure,
					Volumes: []corev1.Volume{
						{
							Name: "cke-certs",
							VolumeSource: corev1.VolumeSource{
								Secret: &corev1.SecretVolumeSource{
									DefaultMode: &Model,
									SecretName:  "cke-certs",
								},
							},
						},
						{
							Name: "ckelogs",
							VolumeSource: corev1.VolumeSource{
								HostPath: &corev1.HostPathVolumeSource{
									Path: constants.AddonsLogDir,
									Type: &constants.MountPathDirType,
								},
							},
						},
					},
				},
			},
		},
	}
}

func RandomUID() string {
	var Len = 6
	var uuid = make([]byte, Len)

	for i := 0; i < Len; i++ {
		uuid[i] = byte(source.Intn(26) + 65 + 32)
	}
	return string(uuid)
}

func ImageGetFromConfig(cfg *ckev1alpha1.CkeClusterSpec) string {
	return GetImage(cfg.Registry.Domain, cfg.Registry.Port, constants.AnsibleImageRepo, cfg.Version, cfg.ClusterParameters.ImageManifestEnabled)
}

func ImageGetFromConfigWithTag(cfg *ckev1alpha1.CkeClusterSpec, tag string) string {
	return GetImage(cfg.Registry.Domain, cfg.Registry.Port, constants.AnsibleImageRepo, tag, cfg.ClusterParameters.ImageManifestEnabled)
}

func GetImage(registry string, port int, name, tag string, disableArchTag bool) string {
	var head, body, end = registry, name, tag

	// if has port, config registry with port as xxxx:xxxx
	if port > 0 {
		head = fmt.Sprintf("%s:%d", registry, port)
	}

	// TODO: delete archtag if has first, makesure old patchCR can use
	// ========================= do this here ? ======================

	// if enable archtag, add arch information for image. such as xxx-amd64
	if !disableArchTag {
		body = fmt.Sprintf("%s-%s", body, constants.ARCH)
	}

	return fmt.Sprintf("%s/%s:%s", head, body, end)
}

func GetNodeByRole(nodes *[]ckev1alpha1.CkeNode, role string) (node *ckev1alpha1.CkeNode) {
	for item := range *nodes {
		for _, nodeRole := range (*nodes)[item].Spec.Roles {
			if role == nodeRole {
				return &(*nodes)[item]
			}
		}
	}
	return nil
}

func Base64Context(cfg *ckev1alpha1.CkeCluster, nodes *[]ckev1alpha1.CkeNode) (base64context string) {
	var out []byte
	var err error
	var content string
	codec, _ := runtime.SerializerInfoForMediaType(serializer.NewCodecFactory(runtime.NewScheme()).SupportedMediaTypes(), "application/yaml")

	if out, err = runtime.Encode(codec.Serializer, cfg); err != nil {
		return ""
	}
	content += fmt.Sprintf("---\n%s\n", string(out))

	for item := range *nodes {
		if out, err = runtime.Encode(codec.Serializer, &(*nodes)[item]); err != nil {
			return ""
		}
		content += fmt.Sprintf("---\n%s\n", string(out))
	}
	return base64.StdEncoding.EncodeToString([]byte(content))
}

func GetPatchList(ckecluster *ckev1alpha1.CkeCluster, ckePatchList *ckev1alpha1.CkePatchList) ([]string, *ckev1alpha1.CkePatch, error) {
	currentVersion := ckecluster.Status.Version // 当前版本
	expectVersion := ckecluster.Spec.Version    // 期望版本

	// 匹配到 to version 则认为可以确定升级路径
	if !checkToVersion(expectVersion, ckePatchList) {
		logrus.Warningf("GetPatchList(): fail to matchs ToVersion from CkePatchCr list")
		return []string{}, &ckev1alpha1.CkePatch{}, fmt.Errorf("can not find expectVersion in ckepatch toversion")
	}

	// 跟据 fromVersion 和 toVersion 选择执行 patch
	ckePatchUse := getUsePatch(currentVersion, expectVersion, ckePatchList)
	if ckePatchUse == nil {
		logrus.Warningf("GetPatchList(): fail to matchs FromVersion from CkePatchCr list")
		return []string{}, &ckev1alpha1.CkePatch{}, fmt.Errorf("can not find currentVersion in ckepatch in fromversion")
	}

	toExecutePatch := Difference(ckePatchUse.Spec.Patches, ckePatchUse.Status.CompletedPatchs)
	return toExecutePatch, ckePatchUse, nil
}

func getUsePatch(fromVersion, toVersion string, patchList *ckev1alpha1.CkePatchList) (usePatch *ckev1alpha1.CkePatch) {
	for item := range patchList.Items {
		if regexp.MustCompile(fmt.Sprintf("^%s$", patchList.Items[item].Spec.FromVersion)).MatchString(fromVersion) && // 匹配到 fromVersion
			version.LessThanOrEqualTo(patchList.Items[item].Spec.ToVersion, toVersion) && (usePatch == nil || version.GreaterThan(patchList.Items[item].Spec.ToVersion, usePatch.Spec.ToVersion)) {
			usePatch = &patchList.Items[item]
		}
	}
	return usePatch
}

func checkToVersion(toVersion string, patchList *ckev1alpha1.CkePatchList) bool {
	for item := range patchList.Items {
		if toVersion == patchList.Items[item].Spec.ToVersion {
			return true
		}
	}
	return false
}

func StrNoLongThan(str string, lens int) string {
	if len(str) <= lens {
		return str
	}

	return str[:lens]
}

func JobFailed(job *batchv1.Job) bool {
	if job.Status.Failed != 0 {
		return true
	}
	if len(job.Status.Conditions) != 0 && job.Status.Conditions[0].Status == corev1.ConditionTrue && job.Status.Conditions[0].Type == batchv1.JobFailed {
		return true
	}
	return false
}

func JobCompleted(job *batchv1.Job) bool {
	if job.Status.Succeeded != 0 {
		return true
	}
	if len(job.Status.Conditions) != 0 && job.Status.Conditions[0].Status == corev1.ConditionTrue && job.Status.Conditions[0].Type == batchv1.JobComplete {
		return true
	}
	return false
}

// return item in sliceA but not in sliceB
func Difference(sliceA, sliceB []string) []string {
	return difference(sliceA, sliceB)
}

// A - B
func difference(sliceA, sliceB []string) (result []string) {
	var keyInB = make(map[string]bool, 0)

	for item := range sliceB {
		keyInB[sliceB[item]] = true
	}

	for item := range sliceA {
		if !keyInB[sliceA[item]] {
			result = append(result, sliceA[item])
		}
	}

	return
}

func SliceEqual(sliceA, sliceB []string) bool {
	return len(difference(sliceA, sliceB)) == 0 && len(difference(sliceB, sliceA)) == 0
}

func AppendIfNotExist(slice []string, appends ...string) (result []string) {
	return append(slice, difference(appends, slice)...)
}

func ContainsStrings(slice []string, s ...string) bool {
	return len(difference(s, slice)) == 0
}

func RemoveStrings(slice []string, s ...string) (result []string) {
	return difference(slice, s)
}

func CheckCkeComponent(toExecutePatchCr ckev1alpha1.CkeComponentPatch, ckecomponent *ckev1alpha1.CkeComponent) (bool, bool) {
	isComponentPatchCompleted := false
	isComponentPatchFailed := false

	if ckecomponent.Status.State == ComponentUpgradeFailedStatus || ckecomponent.Status.State == ComponentUpdateFailedStatus {
		isComponentPatchFailed = true
		return false, isComponentPatchFailed
	}

	if toExecutePatchCr.ToVersion != "" {
		if ckecomponent.Status.State == ComponentRunningStatus && ckecomponent.Status.Version == ckecomponent.Spec.Version && ckecomponent.Status.Version == toExecutePatchCr.ToVersion {
			isComponentPatchCompleted = true
		}
	} else if toExecutePatchCr.ToImageTag != "" {
		if ckecomponent.Status.State == ComponentRunningStatus && ckecomponent.Status.ImageTag == ckecomponent.Spec.ImageTag && ckecomponent.Status.ImageTag == toExecutePatchCr.ToImageTag {
			isComponentPatchCompleted = true
		}
	} else if len(toExecutePatchCr.Parameters) != 0 {
		if ckecomponent.Status.State == ComponentRunningStatus && !IsParametersChanged(ckecomponent) {
			for k, v := range toExecutePatchCr.Parameters {
				if ckecomponent.Status.Parameters[k] != v {
					return isComponentPatchCompleted, isComponentPatchFailed
				}
			}
			isComponentPatchCompleted = true
		}
	}

	return isComponentPatchCompleted, isComponentPatchFailed
}

func IsParametersChanged(ckecomponent *ckev1alpha1.CkeComponent) bool {
	isParametersChanged := false
	specParameters := make(map[string]string)
	if len(ckecomponent.Spec.Parameters) != len(ckecomponent.Status.Parameters) {
		isParametersChanged = true
	} else {
		date, _ := json.Marshal(ckecomponent.Spec.Parameters)
		json.Unmarshal(date, &specParameters)
		for k, v := range ckecomponent.Status.Parameters {
			if specParameters[k] == v {
				delete(specParameters, k)
			} else {
				break
			}
		}
		if len(specParameters) > 0 {
			isParametersChanged = true
		}
	}
	return isParametersChanged
}
